package com.niit;


import org.apache.commons.lang3.RandomStringUtils;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.SimpleDateFormat;
import java.util.*;

public class KafkaProducer {
    public static void main(String[] args) {
        // Kafka配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.136.128:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建Kafka生产者
        Producer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);

        try {
            while (true) {
                // 生成班级号，前四位范围是2020 - 2023，后两位是01 - 05
                String classId = generateClassId();

                String name = RandomStringUtils.randomAlphabetic(5);
                int gender = new Random().nextInt(2);

                // 生成随机出生年月日
                String birth = generateRandomBirthDate();

                // 生成唯一的学号
                String studentId = generateUniqueStudentId();

                int semester = new Random().nextInt(8) + 1;
                char status = (new Random().nextInt(2) == 0)? 'L' : 'A';

                // 拼接数据
                String data = classId + "\t" + name + "\t" + gender + "\t" + birth + "\t" + studentId + "\t" + semester + "\t" + status;

                // 发送数据到Kafka主题
                producer.send(new ProducerRecord<>("stuInfo", data));
                System.out.println("Sent: " + data);

                // 每隔5秒发送一次数据
                Thread.sleep(5000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }

    private static String generateClassId() {
        Random random = new Random();
        // 生成前四位年份，范围是2013 - 2023
        int year = random.nextInt(3) + 2020;
        // 生成后两位班级序号，范围是01 - 05
        int classSeq = random.nextInt(5) + 1;
        return String.format("%04d%02d", year, classSeq);
    }

    private static String generateRandomBirthDate() {
        Random random = new Random();
        // 生成随机年份，范围可根据实际需求调整，这里假设从2000年到2010年
        int year = random.nextInt(10) + 2000;
        // 生成随机月份，范围是1到12
        int month = random.nextInt(12) + 1;
        // 根据月份生成随机日期，考虑不同月份的天数差异
        int day;
        if (month == 2) {
            // 闰年判断
            if ((year % 4 == 0 && year % 100!= 0) || year % 400 == 0) {
                day = random.nextInt(29) + 1;
            } else {
                day = random.nextInt(28) + 1;
            }
        } else if (month == 4 || month == 6 || month == 9 || month == 11) {
            day = random.nextInt(30) + 1;
        } else {
            day = random.nextInt(31) + 1;
        }

        // 将生成的年、月、日格式化为8位数字字符串
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
        Calendar calendar = Calendar.getInstance();
        calendar.set(year, month - 1, day);
        Date date = calendar.getTime();
        return sdf.format(date);
    }

    private static String generateUniqueStudentId() {
        Random random = new Random();
        String studentId;
        Set<String> generatedStudentIds = new HashSet<>();
        do {
            studentId = RandomStringUtils.randomNumeric(8);
        } while (generatedStudentIds.contains(studentId));
        generatedStudentIds.add(studentId);
        return studentId;
    }
}
